package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.app.func.DorisSinkFunction;
import com.atguigu.edu.realtime.beans.CartAddUuBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.DorisUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 加购独立用户数统计
 */
public class DwsTradeCartAddUuWindow {
    public static void main(String[] args) {
        //TODO 1. 基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(5);
        //TODO 2.检查点相关设置(略)
        env.enableCheckpointing(5000L);
        //TODO 3.从kafka主题中读取数据
        //3.1 声明消费主题和消费者组
        String topic="dwd_trade_cart_add";
        String groupId="dws_trade_cart_add_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //TODO 4.对流中的数据进行类型转换 jsonStr ->jsonObj
        SingleOutputStreamOperator<JSONObject> JsonObjDS
                = kafkaStrDS.map(JSON::parseObject);

/*{"course_id": "39","user_id": "47", "course_name": "Java基础","session_id": "ec43d0e0-3377-4088-9e4a-ede734830bf3",
                "id": "11902",
                "ts": "1701056483"}*/
        //JsonObjDS.print(">>>");

        //TODO 5.指定Watermark并提取事件时间字段
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = JsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts")*1000;
                                    }
                                }
                        )
        );

        //TODO 6.按照用户id进行分组
        KeyedStream<JSONObject, String> keyByDS
                = withWatermarkDS.keyBy(jsonObject -> jsonObject.getString("user_id"));

        //TODO 7.使用flink状态编程 判断是否为加购独立用户
        SingleOutputStreamOperator<JSONObject> cartUUDS = keyByDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<String>lastCartDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor
                                = new ValueStateDescriptor<String>("lastCartDateState", String.class);
                        lastCartDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context context, Collector<JSONObject> out) throws Exception {
                        //获取上次加购时间
                        String lastCartDate = lastCartDateState.value();
                       //获取当前加购日期
                        Long ts = jsonObj.getLong("ts")*1000;
                        String curCartDate = DateFormatUtil.toDate(ts);
                        if (StringUtils.isEmpty(lastCartDate)|| !lastCartDate.equals(curCartDate)){
                            out.collect(jsonObj);
                            lastCartDateState.update(curCartDate);
                        }
                    }
                }
        );
        //TODO 8.开窗
        AllWindowedStream<JSONObject, TimeWindow> windowDS
                = cartUUDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));
        //TODO 9.聚合
        SingleOutputStreamOperator<CartAddUuBean> aggregateDS = windowDS.aggregate(
                new AggregateFunction<JSONObject, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(JSONObject value, Long accumulator) {
                        return ++accumulator;
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }
                },
                new AllWindowFunction<Long, CartAddUuBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow Window, Iterable<Long> values, Collector<CartAddUuBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(Window.getStart());
                        String edt = DateFormatUtil.toYmdHms(Window.getEnd());
                        String curDate = DateFormatUtil.toDate(Window.getStart());

                        for (Long value : values) {
                            out.collect(new CartAddUuBean(
                                    stt,
                                    edt,
                                    curDate,
                                    value
                            ));
                        }
                    }
                }
        );
        //TODO 10.将聚合结果写到Doris中
        aggregateDS.print(">>>");
        aggregateDS.addSink(new DorisSinkFunction<CartAddUuBean>("dws_trade_cart_add_uu_window"));
                /*.ma(new BeanToJsonStrFunction<>())
                .sinkTo(DorisUtil.getDorisSink("dws_trade_cart_add_uu_window"));*/

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
